package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
//评价
public class DwdInteractionComment {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
                env.setParallelism(5);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("appraise"));

        Table table = tableEnv.sqlQuery("select `data`['user_id'] user_id,`data`['course_id'] course_id from topic_db where `table`='review_info'");

        tableEnv.createTemporaryView("appraise",table);

        tableEnv.executeSql("CREATE TABLE dwd_appraise (\n" +
                "  user_id string,\n" +
                "  course_id string,\n" +
                "  PRIMARY KEY (user_id) NOT ENFORCED\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_appraise"));
        //5.2 写入
        tableEnv.executeSql("insert into dwd_appraise select * from appraise");

    }
}
